package com.alibaba.nacos.common.remote.client.grpc;

import com.alibaba.nacos.api.exception.NacosException;
import com.alibaba.nacos.api.grpc.auto.Payload;
import com.alibaba.nacos.api.grpc.auto.RequestGrpc;
import com.alibaba.nacos.api.remote.RequestCallBack;
import com.alibaba.nacos.api.remote.RpcScheduledExecutor;
import com.alibaba.nacos.api.remote.request.Request;
import com.alibaba.nacos.api.remote.response.ErrorResponse;
import com.alibaba.nacos.api.remote.response.Response;
import com.alibaba.nacos.api.remote.response.ResponseCode;
import com.alibaba.nacos.common.remote.client.Connection;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import io.grpc.ManagedChannel;
import io.grpc.stub.StreamObserver;
import lombok.Setter;
import lombok.SneakyThrows;
import lombok.experimental.Accessors;

import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;


@Accessors(chain = true)
@Setter
public class GrpcConnection
        extends Connection {

    protected ManagedChannel channel;
    private Executor executor;
    protected RequestGrpc.RequestFutureStub requestFutureStub;

    @Override
    public void asyncRequest(Request request,
                             RequestCallBack requestCallBack) throws NacosException {
        Payload payload = GrpcUtils.convert(request);
        ListenableFuture<Payload> requestFuture = requestFutureStub.request(payload);
        Futures.addCallback(requestFuture, new FutureCallback<>() {
            @Override
            public void onSuccess(Payload result) {
                Response response = GrpcUtils.parse(result);
                if (response == null) {
                    requestCallBack.onException(new NacosException(ResponseCode.FAIL.getCode(), "response is null"));
                } else if (response instanceof ErrorResponse errorResponse) {
                    requestCallBack.onException(new NacosException(errorResponse.getErrorCode(), errorResponse.getMessage()));
                } else {
                    requestCallBack.onResponse(response);
                }
            }

            @Override
            public void onFailure(Throwable t) {
                throw new UnsupportedOperationException(t);
            }
        }, executor);

        Futures.withTimeout(requestFuture, requestCallBack.getTimeout(), TimeUnit.MILLISECONDS, RpcScheduledExecutor.TIMEOUT_SCHEDULER);
    }

    @SneakyThrows
    @Override
    public Response request(Request request,
                            long timeoutLills) throws NacosException {
        Payload requestPayload = GrpcUtils.convert(request);
        ListenableFuture<Payload> future = requestFutureStub.request(requestPayload);
        Payload responsePayload = future.get(timeoutLills, TimeUnit.MILLISECONDS);
        return GrpcUtils.parse(responsePayload);
    }
}
